package com.xujian.cdc.broadcast;

import com.alibaba.fastjson2.JSONObject;
import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import com.xujian.cdc.schema.JsonDebeziumDeserializationSchema;
import com.xujian.common.beans.JsonRecord;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author star xu
 * @date 2023/1/4 16:08
 * @Description:广播
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class BroadCastExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> packageIdStram = env.fromElements("12345", "1234", "YZ2021");
        Map<String, String> packageMap = new HashMap<>();

        //维表
        String tableName = "bsae_package";
        String srcHost = "localhost";
        String srcDatabase = "whe_member_meinian";
        String srcUsername = "postgres";
        String srcPassword = "qwer1234";
        String schemaName = "public";
        Properties properties = new Properties();
        properties.setProperty("slot.name", "flink_rel");

        SourceFunction<JsonRecord> pgSqlSource = PostgreSQLSource.<JsonRecord>builder()
                .hostname(srcHost)
                .port(5432)
                .database(srcDatabase)
                .schemaList(schemaName)
                .tableList(schemaName + "." + tableName)
                .username(srcUsername)
                .password(srcPassword)
                .decodingPluginName("pgoutput")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .debeziumProperties(properties)
                .build();

        final DataStreamSource<JsonRecord> dimSource = env.addSource(pgSqlSource);

        final MapStateDescriptor<String, String> dim_bsae_package_broadcast = new MapStateDescriptor<>("dim_bsae_package_broadcast", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
        final BroadcastStream<JsonRecord> dimBc = dimSource.broadcast(dim_bsae_package_broadcast);
        BroadcastConnectedStream<String, JsonRecord> connectedStreams = packageIdStram.connect(dimBc);

        SingleOutputStreamOperator<String> map = connectedStreams.process(new BroadcastProcessFunction<String, JsonRecord, String>() {
            MapStateDescriptor<String, String> mapStateDescriptor;
            final Map<String, String> initMap = new HashMap<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //初始化维表数据
                Connection connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/whe_member_meinian", "postgres", "qwer1234");
                PreparedStatement preparedStatement = connection.prepareStatement("select package_id,package_name from bsae_package");
                ResultSet resultSet = preparedStatement.executeQuery();
                ResultSetMetaData metaData = resultSet.getMetaData();
                while (resultSet.next()) {
                    initMap.put(resultSet.getString(1), resultSet.getString(2));
                }
                resultSet.close();
                preparedStatement.close();
                connection.close();
                if (mapStateDescriptor == null) {
                    mapStateDescriptor = new MapStateDescriptor<>("dim_bsae_package_broadcast", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
                }
            }

            @Override
            public void processElement(String value, BroadcastProcessFunction<String, JsonRecord, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                final ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("package_id", value);
                if (broadcastState.contains(value)) {
                    jsonObject.put("package_name", broadcastState.get(value));
                } else {
                    jsonObject.put("package_name", StringUtils.isNotBlank(initMap.get(value)) ? initMap.get(value) : "未知");
                }
                out.collect(jsonObject.toJSONString());
            }

            @Override
            public void processBroadcastElement(JsonRecord value, BroadcastProcessFunction<String, JsonRecord, String>.Context ctx, Collector<String> collector) throws Exception {
                final BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                final JSONObject jsonObject = JSONObject.parseObject(value.getFieldValue());
                String package_id = jsonObject.getString("package_id");
                switch (value.getOp()) {
                    case "d":
                        System.out.println("移除: " + package_id);
                        broadcastState.remove(package_id);
                        break;
                    case "i":
                    case "u":
                    default:
                        broadcastState.put(package_id, jsonObject.getString("package_name"));
                }
            }
        });

        map.print();
        env.execute();
    }
}
